"""
Case Type   : fullsql
Case Name   : full sql信息清理验证(full函数查询:1, slow函数查询:1)
Create At   : 2022/01/26
Owner       : zhao-pengcheng-g
Description :
    1.检查参数，修改配置:enable_copy_server_files=on
    2.检查参数，修改配置:instr_unique_sql_count=100000
    3.创建用户，创建表
    4.导入数据
    5.设置track_stmt_session_slot为1000
    6.设置set track_stmt_stat_level和log_min_duration_statement的值,
    执行大于log_min_duration_statement阈值的query
    7.full sql函数查询信息
    8.slow sql函数查询信息
    9.statement_detail_decode校验
    10.环境清理
Expect      :
    1.修改参数成功
    2.修改参数成功
    3.创建成功
    4.导入成功
    5.设置成功
    6.设置成功,执行query成功
    7.成功，1条记录
    8.成功，1条记录
    9.成功
    10.环境清理成功
History     :
"""

import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common


class Fullsql(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        text = '---Opengauss_Function_Fullsql_Case0005 start---'
        self.logger.info(text)
        self.userNode = Node('PrimaryDbUser')
        self.rootNode = Node('PrimaryRoot')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.table_name_1 = 't_part_fullsql_0005_1'
        self.table_name_2 = 't_lineitem_fullsql_0005_2'
        self.user_name = 'u_fullsql_0005'
        self.table_path = os.path.join(r'/home',
                                       self.userNode.ssh_user, 'copy_tables')

        text = '---step1:检查参数，修改配置:enable_copy_server_files=on,' \
               'expect:成功---'
        self.logger.info(text)
        self.config_copy = 'enable_copy_server_files'
        self.show_copy_cmd = f'show {self.config_copy};'
        check_copy = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(check_copy)
        self.copy_default = check_copy.splitlines()[-2].strip()
        if 'on' != self.copy_default:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_copy}=on')
            self.assertTrue(result, '参数修改失败')
        chk_copy_msg = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(chk_copy_msg)
        self.common.equal_sql_mdg(chk_copy_msg, self.config_copy, 'on',
                                  '(1 row)', flag='1')
        text = '---step2:检查参数，修改配置:instr_unique_sql_count=100000,' \
               'expect:成功---'
        self.logger.info(text)
        self.config_instr = 'instr_unique_sql_count'
        self.show_instr_cmd = f'show {self.config_instr};'
        check_instr = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(check_instr)
        self.instr_default = check_instr.splitlines()[-2].strip()
        if '100000' != self.instr_default:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_instr}=100000')
            self.assertTrue(result, '参数修改失败')
        chk_instr_msg = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(chk_instr_msg)
        self.common.equal_sql_mdg(chk_instr_msg, self.config_instr, '100000',
                                  '(1 row)', flag='1')

        text = '---step3:清理历史记录,创建用户，创建表,expect;成功---'
        self.logger.info(text)
        sql_cmd = f'''select reset_unique_sql('global', 'all', 1);
                        drop user if exists {self.user_name} cascade;
                        create user {self.user_name} sysadmin \
                        password '{macro.COMMON_PASSWD}'; '''
        self.logger.info(sql_cmd)
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd)
        self.logger.info(sql_msg)
        self.assertIn('reset_unique_sql' and '(1 row)',
                      sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)

        text = '---step3.1:切换用户,创建part表,expect:创建成功 ---'
        self.logger.info(text)
        create_part = f'''set role {self.user_name} \
                    password '{macro.COMMON_PASSWD}';
                    set current_schema to public;
                    drop table if exists {self.table_name_1};
                    create table {self.table_name_1}
                    (
                    p_partkey     bigint not null, 
                    p_name        varchar(55) not null, 
                    p_mfgr        char(25) not null, 
                    p_brand       char(10) not null, 
                    p_type        varchar(25) not null, 
                    p_size        bigint not null, 
                    p_container   char(10) not null, 
                    p_retailprice decimal(15,2) not null, 
                    p_comment     varchar(23) not null
                    )
                    partition by range(p_size)
                    (
                    partition p_size_1 values less than(11),
                    partition p_size_2 values less than(21),
                    partition p_size_3 values less than(31),
                    partition p_size_4 values less than(41),
                    partition p_size_5 values less than(51)
                    );'''
        msg_part = self.sh_primy.execut_db_sql(create_part)
        self.logger.info(msg_part)
        self.assertTrue(msg_part.count(self.constant.SET_SUCCESS_MSG) == 2,
                        '执行失败:' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS, msg_part,
                      '执行失败:' + text)

        text = '---step3.2:切换用户,创建lineitem表,expect:创建成功 ---'
        self.logger.info(text)
        create_lineitem = f'''set role {self.user_name} \
            password '{macro.COMMON_PASSWD}';
            set current_schema to public;
            drop table if exists {self.table_name_2};
            create table {self.table_name_2}
            (
            l_orderkey    bigint not null, 
            l_partkey     bigint not null, 
            l_suppkey     bigint not null, 
            l_linenumber  bigint not null, 
            l_quantity    decimal(15,2) not null, 
            l_extendedprice  decimal(15,2) not null, 
            l_discount    decimal(15,2) not null, 
            l_tax         decimal(15,2) not null, 
            l_returnflag  char(1) not null, 
            l_linestatus  char(1) not null, 
            l_shipdate    date not null, 
            l_commitdate  date not null, 
            l_receiptdate date not null, 
            l_shipinstruct char(25) not null, 
            l_shipmode     char(10) not null, 
            l_comment      varchar(44) not null
            )
            partition by range(l_shipdate)
            (
            partition l_shipdate_1 values less than('1993-01-01 00:00:00'),
            partition l_shipdate_2 values less than('1994-01-01 00:00:00'),
            partition l_shipdate_3 values less than('1995-01-01 00:00:00'),
            partition l_shipdate_4 values less than('1996-01-01 00:00:00'),
            partition l_shipdate_5 values less than('1997-01-01 00:00:00'),
            partition l_shipdate_6 values less than('1998-01-01 00:00:00'),
            partition l_shipdate_7 values less than('1999-01-01 00:00:00')
            ); '''
        msg_lineitem = self.sh_primy.execut_db_sql(create_lineitem)
        self.logger.info(msg_lineitem)
        self.assertTrue(msg_lineitem.count(self.constant.SET_SUCCESS_MSG) == 2,
                        '执行失败:' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS, msg_lineitem,
                      '执行失败:' + text)

        text = '---step4:导入数据，expect:导入成功---'
        self.logger.info(text)
        mkdir_cmd = f'rm -rf {self.table_path};' \
                    f'mkdir {self.table_path}'
        self.userNode.sh(mkdir_cmd)
        self.common.scp_file(self.userNode, 'part.tbl',
                             self.table_path)
        self.common.scp_file(self.userNode, 'lineitem.tbl',
                             self.table_path)
        part_path = os.path.join(self.table_path, 'part.tbl')
        lineitem_path = os.path.join(self.table_path, 'lineitem.tbl')
        copy_cmd = f'''set role {self.user_name} \
                    password '{macro.COMMON_PASSWD}';
                    copy {self.table_name_1} from '{part_path}' \
                    using delimiters '|';
                    copy {self.table_name_2} from '{lineitem_path}' \
                    using delimiters '|';
                    analyze;'''
        self.logger.info(copy_cmd)
        copy_msg = self.sh_primy.execut_db_sql(copy_cmd)
        self.logger.info(copy_msg)
        self.assertTrue(copy_msg.count(self.constant.copy_success_msg) == 2,
                        '执行失败:' + text)
        self.assertIn(self.constant.SET_SUCCESS_MSG
                      and self.constant.ANALYZE_SUCCESS_MSG,
                      copy_msg, '执行失败:' + text)

    def test_fullsql(self):
        text = '---step5_1:设置track_stmt_session_slot为1000,expect:设置成功 ---'
        self.logger.info(text)
        self.config_slot = 'track_stmt_session_slot'
        self.show_slot_cmd = f'show {self.config_slot};'
        check_slot = self.sh_primy.execut_db_sql(self.show_slot_cmd)
        self.logger.info(check_slot)
        self.slot_default = check_slot.splitlines()[-2].strip()
        if '1000' != self.slot_default:
            guc_slot = self.sh_primy.execute_gsguc('reload',
                                                   self.constant.GSGUC_SUCCESS_MSG,
                                                   f'{self.config_slot}=1000')
            self.assertTrue(guc_slot, '执行失败:' + text)
        chk_slot_msg = self.sh_primy.execut_db_sql(self.show_slot_cmd)
        self.logger.info(chk_slot_msg)
        self.common.equal_sql_mdg(chk_slot_msg, self.config_slot, '1000',
                                  '(1 row)', flag='1')
        text = '---step5_2:设置full sql 信息清理为5s,expect:设置成功---'
        self.logger.info(text)
        self.config_time = 'track_stmt_retention_time'
        self.show_time_cmd = f'show {self.config_time};'
        check_time = self.sh_primy.execut_db_sql(self.show_time_cmd)
        self.logger.info(check_time)
        self.time_default = check_time.splitlines()[-2].strip()
        if '5,604800' != self.time_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                                     self.constant.GSGUC_SUCCESS_MSG,
                                                     f"{self.config_time}='5,604800'")
            self.assertTrue(guc_result, '执行失败:' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_time_cmd)
        self.logger.info(show_msg)
        self.assertIn('5,604800', show_msg, '执行失败:' + text)

        text = '---step6:设置set track_stmt_stat_level和' \
               'log_min_duration_statement的值,' \
               '执行大于log_min_duration_statement阈值的query,' \
               'expect:设置成功，执行query成功---'
        self.logger.info(text)
        select_cmd = f'''set role {self.user_name} \
                password '{macro.COMMON_PASSWD}';
                set track_stmt_stat_level='l1,l1';
                set log_min_duration_statement='0';
                show log_min_duration_statement;
                show track_stmt_stat_level;
                select 100 * sum(case 
                    when p_type like 'PROMO%'
                        then l_extendedprice * (1 - l_discount)/10
                        else 0
                end) / sum(l_extendedprice * (1 - l_discount)/1000) 
                as promo_order_count
                from {self.table_name_2},{self.table_name_1} 
                where l_partkey = p_partkey
                    and l_shipdate >= date '1995-09-01'
                    and l_shipdate < date '1995-09-01' + interval '1 month';
                select pg_sleep(60);'''
        select_msg = self.sh_primy.execut_db_sql(select_cmd)
        self.logger.info(select_msg)
        set_assert = select_msg.count(self.constant.SET_SUCCESS_MSG) == 3
        self.assertTrue(set_assert, '执行失败:' + text)
        self.assertIn('0', select_msg.splitlines()[5], '执行失败:' + text)
        self.assertIn('l1,l1', select_msg.splitlines()[10], '执行失败:' + text)
        self.assertIn('1638.07786263955401479927',
                      select_msg, '执行失败:' + text)

        text = '---step7: full sql函数查询信息,expect:成功，1条记录---'
        self.logger.info(text)
        fullsql_cmd = f'''select * 
            from dbe_perf.get_global_full_sql_by_timestamp\
                (clock_timestamp() - 1/24,clock_timestamp()) 
            where unique_query_id=(select unique_sql_id 
                from dbe_perf.statement 
                where query like '%when p_type like%');'''
        fullsql_msg = self.sh_primy.execut_db_sql(fullsql_cmd,
                                                  dbname='postgres')
        self.logger.info(fullsql_msg)
        self.assertIn('(1 row)', fullsql_msg, '执行失败:' + text)

        text = '---step8:slow sql函数查询信息,expect:1条记录---'
        self.logger.info(text)
        slowsql_cmd = f'''select * from 
                dbe_perf.get_global_slow_sql_by_timestamp\
                (clock_timestamp() - 1/24, clock_timestamp()) 
            where unique_query_id=(select unique_sql_id 
                from dbe_perf.statement 
                where query like '%when p_type like%' 
                    and user_name = '{self.user_name}');'''
        slowsql_msg = self.sh_primy.execut_db_sql(slowsql_cmd,
                                                  dbname='postgres')
        self.logger.info(slowsql_msg)
        self.assertIn('(1 row)', slowsql_msg, '执行失败:' + text)

        text = "---step9:statement_detail_decode校验，expect:成功---"
        self.logger.info(text)
        decode_cmd = f'''select 
                statement_detail_decode(details, 'plaintext', true)  
            from statement_history 
            where unique_query_id=(select unique_sql_id 
                from dbe_perf.statement 
                where query like '%when p_type like%' 
                    and user_name = '{self.user_name}');'''
        decode_msg = self.sh_primy.execut_db_sql(decode_cmd,
                                                 dbname='postgres')
        self.logger.info(decode_msg)
        self.assertNotIn(self.constant.TPCC_ERROR,
                         decode_msg, '执行失败' + text)
        self.assertIn('statement_detail_decode' and '(1 row)',
                      decode_msg, '执行失败:' + text)

    def tearDown(self):
        text = '---step10.环境清理，expect:成功---'
        self.logger.info(text)
        text = '---step10.1检查参数，恢复原值:enable_copy_server_files,' \
               'expect:成功---'
        self.logger.info(text)
        check_copy = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(check_copy)
        current = check_copy.splitlines()[-2].strip()
        if self.copy_default != current:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_copy}={self.copy_default}')
            self.assertTrue(result, '参数恢复失败')
        show_copy_msg = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(show_copy_msg)

        text = '---step10.2检查参数，恢复原值:instr_unique_sql_count,' \
               'expect:成功---'
        self.logger.info(text)
        check_instr = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(check_instr)
        current = check_instr.splitlines()[-2].strip()
        if self.instr_default != current:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_instr}={self.instr_default}')
            self.assertTrue(result, '参数恢复失败')
        show_instr_msg = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(show_instr_msg)

        text = '---step10.3检查参数，恢复原值:track_stmt_session_slot,' \
               'expect:成功---'
        self.logger.info(text)
        check_slot = self.sh_primy.execut_db_sql(self.show_slot_cmd)
        self.logger.info(check_slot)
        current = check_slot.splitlines()[-2].strip()
        if self.slot_default != current:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_slot}={self.slot_default}')
            self.assertTrue(result, '参数恢复失败')
        show_slot_msg = self.sh_primy.execut_db_sql(self.show_slot_cmd)
        self.logger.info(show_slot_msg)

        text = '---step13.4:检查参数，恢复原值:track_stmt_retention_time，' \
               'expect:成功---'
        self.logger.info(text)
        check_time = self.sh_primy.execut_db_sql(self.show_time_cmd)
        self.logger.info(check_time)
        current = check_time.splitlines()[-2].strip()
        if self.time_default != current:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f"{self.config_time}='{self.time_default}'")
            self.assertTrue(result, '参数恢复失败')
        show_time_msg = self.sh_primy.execut_db_sql(self.show_time_cmd)
        self.logger.info(show_time_msg)

        text = '---step10.4删除表，删除用户,删除文件夹,expect:成功---'
        self.logger.info(text)
        drop_cmd = f'''drop owned by {self.user_name} cascade;
                      drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd)
        self.logger.info(drop_msg)
        rm_cmd = f'rm -rf {self.table_path};'
        self.rootNode.sh(rm_cmd)
        self.assertIn(self.constant.DROP_OWNED_SUCCESS and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败:' + text)
        self.assertFalse(os.path.exists(self.table_path), '执行失败:' + text)
        self.common.equal_sql_mdg(show_copy_msg, self.config_copy,
                                  self.copy_default, '(1 row)', flag='1')
        self.common.equal_sql_mdg(show_instr_msg, self.config_instr,
                                  self.instr_default, '(1 row)', flag='1')
        self.common.equal_sql_mdg(show_slot_msg, self.config_slot,
                                  self.slot_default, '(1 row)', flag='1')
        self.common.equal_sql_mdg(show_time_msg, self.config_time,
                                  self.time_default, '(1 row)', flag='1')
        self.logger.info(
            '---Opengauss_Function_Fullsql_Case0005 finish---')
